package org.xx.armory.db.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xx.armory.db.ConnectionRef;
import org.xx.armory.db.DbException;
import org.xx.armory.db.SqlAction;
import org.xx.armory.db.Transaction;
import org.xx.armory.db.TransactionProperties;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import static java.sql.Connection.TRANSACTION_SERIALIZABLE;
import static org.apache.commons.lang3.StringUtils.join;
import static org.xx.armory.commons.Validators.notNull;

/**
 * 基础的事务实现。
 *
 * @author Haart
 */
abstract class AbstractTransaction
        implements Transaction, AutoCloseable {
    private final Logger logger = LoggerFactory.getLogger(AbstractTransaction.class);
    private final Set<Object> users;
    private final ConnectionRef connection;
    private final TransactionProperties properties;
    private final AtomicBoolean dirty;
    private final AtomicBoolean active;

    /**
     * 构造{@code AbstractTransaction}实例。
     *
     * @param dataSource
     *         事务使用的数据源。
     * @param properties
     *         事务的属性。
     */
    protected AbstractTransaction(
            DataSource dataSource,
            TransactionProperties properties
    ) {
        notNull(dataSource, "dataSource");

        this.users = new HashSet<>();
        this.connection = new ConnectionWrapper(dataSource, false, false,
                                                properties.isSerializable() ? TRANSACTION_SERIALIZABLE : Connection.TRANSACTION_READ_COMMITTED);
        this.properties = notNull(properties, "properties");
        this.dirty = new AtomicBoolean(false);
        this.active = new AtomicBoolean(true);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized void close()
            throws Exception {
        try {
            this.connection.close();
        } finally {
            this.active.set(false);
        }
    }

    /**
     * 获取当前事务关联的数据库连接引用。
     *
     * @return 当前事务关联的数据库连接引用。
     */
    protected final ConnectionRef getConnection() {
        return this.connection;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final ConnectionRef enroll(
            Object user
    ) {
        notNull(user, "user");

        synchronized (this.users) {
            if (doEnroll(this.users, user)) {
                logger.trace(this.toString() + " enroll " + user);
            }
        }

        return getConnection();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void revoke(
            Object user
    ) {
        notNull(user, "user");

        synchronized (this.users) {
            if (doRevoke(this.users, user)) {
                logger.trace(this.toString() + " revoke " + user);

                if (this.getProperties().isAutoCommit()) {
                    commit();
                }
            }
        }
    }
//
//    /**
//     * 设置当前事务使用的数据库连接引用。
//     *
//     * @param connection
//     *         当前事务使用的数据库连接引用。
//     * @throws IllegalStateException
//     *         如果当前事务已初始化。
//     * @throws IllegalArgumentException
//     *         如果参数{@code connection}是{@code null}。
//     */
//    public final void setConnection(
//            ConnectionRef connection
//    ) {
//        assertNotInitialized();
//
//        this.connection = notNull(connection, "connection");
//    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final Set<Object> getUsers() {
        synchronized (this.users) {
            return Collections.unmodifiableSet(this.users);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void setDirty() {
        this.dirty.set(true);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void commit() {
        try {
            this.connection.execute(connection -> {
                if (this.properties.isReadOnly()) {
                    // 对于只读事务，用回滚代替提交。
                    if (this.dirty.compareAndSet(true, false)) {
                        connection.rollback();
                        logger.trace("rollback readonly transaction, connection = {}", connection);
                    } else {
                        logger.trace("transaction is not dirty, skip rolling");
                    }
                } else {
                    if (this.dirty.compareAndSet(true, false)) {
                        connection.commit();
                        logger.trace("commit transaction, connection = {}", connection);
                    } else {
                        logger.trace("transaction is not dirty, skip committing");
                    }
                }
                return null;
            });
        } catch (SQLException ex) {
            throw new DbException(ex);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void rollback() {
        try {
            this.connection.execute(connection -> {
                if (this.dirty.compareAndSet(true, false)) {
                    connection.rollback();
                    logger.trace("rollback transaction, connection = {}", connection);
                } else {
                    logger.trace("transaction is not dirty, skip rolling");
                }
                return null;
            });
        } catch (SQLException ex) {
            throw new DbException(ex);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isActive() {
        return this.active.get();
    }

    protected boolean doEnroll(
            Set<Object> users,
            Object user
    ) {
        return users.add(user);
    }

    protected boolean doRevoke(
            Set<Object> users,
            Object user
    ) {
        return users.remove(user);
    }

    /**
     * 获取当前事务的属性。
     *
     * @return 当前事务的属性。
     */
    protected final TransactionProperties getProperties() {
        return this.properties;
    }

//    /**
//     * 设置当前事务使用的属性。
//     *
//     * @param properties
//     *         当前事务使用的属性。
//     * @throws IllegalStateException
//     *         如果当前事务已初始化。
//     * @throws IllegalArgumentException
//     *         如果参数{@code properties}是{@code null}。
//     */
//    public final void setProperties(
//            TransactionProperties properties
//    ) {
//        assertNotInitialized();
//
//        this.properties = notNull(properties, "properties");
//    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final String toString() {
        return super.toString() + "{users=[" + join(this.users, ",") + "]}";
    }

    /**
     * 内部的数据库连接引用的实现。
     */
    private static final class ConnectionWrapper
            implements ConnectionRef {
        private final Logger logger = LoggerFactory.getLogger(ConnectionWrapper.class);

        private final DataSource dataSource;
        private final boolean autoCommit;
        private final boolean readOnly;
        private final int isolationLevel;
        private Connection connection;

        /**
         * 创建内部的数据连接引用的实现。
         *
         * @param dataSource
         *         用于创建实际数据库连接的数据源。
         * @param autoCommit
         *         是否自动提交。
         * @param readOnly
         *         是否只读。
         * @param isolationLevel
         *         隔离级别。
         * @throws IllegalArgumentException
         *         如果参数{@code dataSource}或者{@code properties}是{@code null}。
         */
        ConnectionWrapper(
                DataSource dataSource,
                boolean autoCommit,
                boolean readOnly,
                int isolationLevel
        ) {
            this.dataSource = notNull(dataSource, "dataSource");
            this.autoCommit = autoCommit;
            this.readOnly = readOnly;
            this.isolationLevel = isolationLevel;
            this.connection = null;
        }

        /**
         * 判断指定的连接是否可用。
         *
         * @param connection
         *         指定的数据库连接。
         * @return 如果可用则返回{@code true}，否则返回{@code false}。
         * @throws SQLException
         *         如果判断数据库连接状态是出错。
         */
        private static boolean isAlive(
                Connection connection
        )
                throws SQLException {
            return connection != null && !connection.isClosed();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public final synchronized <R> R execute(
                SqlAction<R> action
        )
                throws SQLException {
            Connection connection = this.connection;
            if (!isAlive(connection)) {
                connection = this.dataSource.getConnection();
                connection.setAutoCommit(this.autoCommit);
                connection.setReadOnly(this.readOnly);
                connection.setTransactionIsolation(this.isolationLevel);
                this.connection = connection;
                logger.trace("open connection: {}", connection);
            }
            return action.call(connection);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public final synchronized void close()
                throws SQLException {
            Connection connection = this.connection;
            if (isAlive(connection)) {
                // 尝试回滚未提交的事务。
                try {
                    connection.rollback();
                } catch (SQLException ignore) {
                }
                connection.close();
                this.connection = null;
                logger.trace("close connection: {}", connection);
            }
        }
    }
}
